#include "catch.hpp"

#include "messaging/rpc/client.h"
#include "messaging/rpc/server.h"
#include "messaging/endpoint.h"
#include "common/path.h"
#include "common/timeit.h"
#include "common/scope_guard.h"


using namespace std;
using namespace evm;
using namespace evm::messaging;
using namespace evm::config;

struct Echo : Endpoint {
    Echo() {
        target = Target::endpoint_target();

        reg("echo", &Echo::echo);
        reg("print", &Echo::print);
    }

    Json echo(const Json& args) {
        log_info("echo: {}", args.dump());
        ++count;
        return Json(args);
    }

    void print(const Json& args) {
        ++count;
        log_info("print: {}", args.dump());
    }

    static atomic<size_t> count;
};

atomic<size_t> Echo::count(0);

struct Resource : Endpoint {
    Resource() {
        target = Target::endpoint_target("resources");

        reg("create", &Resource::create);
        reg("show", &Resource::show);
        reg("delete", &Resource::delete_);
    }

    Json create(const Json& args) {
        log_info("create: {}", args.dump());
        if (args.count("resource"))
            resource = args["resource"].get<string>();
        return {{"resource", resource}};
    }

    Json show(const Json& args) {
        log_info("show: {}", args.dump());
        return {{"resource", resource}};
    }

    void delete_(const Json& args) {
        log_info("delete_: {}", args.dump());
        resource.clear();
    }

    string resource;
};

TEST_CASE("messaging rpc tests", "[messaging_rpc]") {


    string conf_file = "messaging.json";
    REQUIRE(path::exists(conf_file));

    auto conf = make_shared<ConfigOpts>("rpc_tests", conf_file);
    register_opts(conf);

    //conf->override_opt("rpc_redis_address", "10.141.123.176:6379");

    (*conf)();

    conf->log();

    SECTION("Single server, single client") {

        auto transport = get_transport(conf);

        auto server_target = Target::server_target("test_topic", "test_server");

        Endpoints endpoints;
        endpoints.emplace_back(new Echo());
        endpoints.emplace_back(new Resource());

        auto server = get_rpc_server(transport, server_target, endpoints);
        server->start();
        log_info("rpc server started");

        try {
            auto client_target = Target::client_target("test_topic");
            auto client = make_shared<RPCClient>(transport, client_target, 0, 3);

            Json args = {{"hello", "world"}};
            Json res;

            res = client->call("echo", args);
            REQUIRE(res == args);

            res = client->call("print", args);
            REQUIRE(res.is_null());

            client->cast("print", args);

            REQUIRE_THROWS_AS(client->call("create", {{"resource", "abcdefg"}}), CallError);

            auto resource_client = client->prepare({{"namespace", "resources"}});
            res = resource_client->call("create", {{"resource", "abcdefg"}});
            REQUIRE(res == Json({{"resource", "abcdefg"}}));
        }
        catch (const exception &e) {
            log_error(e.what());
        }

        server->stop();
        server->wait();

        log_info("rpc_tests stopped");
    }

    SECTION("Mutiple servers with same topic, mutiple clients") {

        int num_servers = 2;
        int num_clients = 5;

        const char* topic = "test_topic";

        Endpoints endpoints;
        endpoints.emplace_back(new Echo());
        endpoints.emplace_back(new Resource());



        vector<MessageServerPtr> servers;
        for (int i = 0; i < num_servers; ++i) {
            auto transport = get_transport(conf);
            auto target = Target::server_target(topic);
            auto server = get_rpc_server(transport, target, endpoints);
            server->start();
            servers.push_back(server);
        }

        SCOPE_EXIT {
            vector<thread> threads;
            for (auto &server: servers) {
               server->stop();
               threads.emplace_back([server] { server->wait(); });
            }
            for (auto &t: threads)
               t.join();
        };

        vector<RPCClientPtr> clients;
        for (int i = 0; i < num_clients; ++i) {
            auto transport = get_transport(conf);
            auto target = Target::client_target(topic);
            auto client = make_shared<RPCClient>(transport, target, 0, 3);
            clients.push_back(client);
        }

        log_info("==================== call ====================");

        Json args = {{"hello", "world"}};

        int count = 1000;

        Echo::count = 0;

        try {
            for (int i = 0; i < count; ++i) {
                for (auto &client: clients)
                    client->call("echo", args);
            }
        } catch (const exception& e) {
            log_error(e.what());
        }

        REQUIRE(count * num_clients <= Echo::count);  // consider extra retry calls

        log_info("==================== cast ====================");

        Echo::count = 0;

        for (int i = 0; i < count; ++i) {
            for (auto& client: clients)
                client->cast("print", args);
        }

        while (count * num_clients - Echo::count > 100) {
            evm::sleep(2);
        }
        evm::sleep(2);

        REQUIRE(count * num_clients <= Echo::count);
    }
}
